package com.firmiana.jt809.netty.sub;

import com.firmiana.jt809.common.ConnectionStateEnum;
import com.firmiana.jt809.netty.JT809MessageDecoder;
import com.firmiana.jt809.netty.JT809MessageEncoder;
import com.firmiana.jt809.service.AbstractBaseService;
import com.firmiana.jt809.service.Listener;
import com.firmiana.jt809.service.ServiceException;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import lombok.extern.slf4j.Slf4j;

import java.net.InetSocketAddress;
import java.util.concurrent.atomic.AtomicReference;

/**
 * 809子链路
 * 上级平台向下级平台发送数据
 *
 * @author hucl
 * @date 2020/8/25 14:19
 */
@Slf4j
public class NettyTCPSubLink extends AbstractBaseService {
    protected final AtomicReference<ConnectionStateEnum> clientState = new AtomicReference<>(ConnectionStateEnum.Created);

    private EventLoopGroup workerGroup;
    protected Bootstrap bootstrap;
    private ChannelFuture channelFuture;
    private Channel channel;
    private String host;
    private int port;

    public NettyTCPSubLink(String host, int port) {
        this.host = host;
        this.port = port;
    }

    public NettyTCPSubLink() {
    }

    @Override
    public void init() {
        if (!clientState.compareAndSet(ConnectionStateEnum.Created, ConnectionStateEnum.Initialized)) {
            throw new ServiceException("sub link already init");
        }
    }

    @Override
    public void start(Listener listener) {
        if (!clientState.compareAndSet(ConnectionStateEnum.Initialized, ConnectionStateEnum.Starting)) {
            throw new ServiceException("sub link already started or have not init");
        }
        createNioClient(listener);
        connect(listener);
    }

    private void createNioClient(Listener listener) {
        NioEventLoopGroup workerGroup = new NioEventLoopGroup(getWorkThreadNum());
        createClient(listener, workerGroup);
    }

    private void createClient(Listener listener, EventLoopGroup workerGroup) {
        this.workerGroup = workerGroup;
        this.bootstrap = new Bootstrap();
        try {
            bootstrap.group(workerGroup)
//                    .option(ChannelOption.SO_REUSEADDR, true)//
//                    .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
                    .option(ChannelOption.SO_KEEPALIVE, true)
                    .channel(NioSocketChannel.class)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            initPipeline(ch.pipeline());
                        }
                    });
        } catch (Exception e) {
            clientState.set(ConnectionStateEnum.Shutdown);
            e.printStackTrace();
        }

    }

    @Override
    public boolean sendMessage(Object msg) {
        try {
            if (channel == null || !channel.isActive()) {
                this.connect(null);
            }
            channel.writeAndFlush(msg);
            return true;
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
    }


    @Override
    public boolean isRunning() {
        return clientState.get() == ConnectionStateEnum.Started;
    }

    private void connect(Listener listener) {
        InetSocketAddress socketAddress = new InetSocketAddress(host, port);
        try {
            if (clientState.get() == ConnectionStateEnum.Starting) {
                channelFuture = bootstrap.connect(socketAddress).addListener(future -> {
                    if (future.isSuccess()) {
                        log.info("子链路启动成功...");
                        clientState.set(ConnectionStateEnum.Started);
                        if (listener != null) {
                            listener.onSuccess(port, host);
                        }
                    } else {
                        log.error("子链路启动失败,port:{}, address:{}, err msg:{}", port, host, future.cause());
                        if (listener != null) {
                            listener.onFailure(future.cause());
                        }
//                    stop(listener);
                    }
                }).sync();
                channel = channelFuture.channel();
            }
        } catch (Exception e) {
            clientState.set(ConnectionStateEnum.Shutdown);
            log.error("子链路启动失败，", e.getCause());
        }
    }

    /**
     * 真对已管理的子链路 重新连接
     */
    @Override
    public void reconnect() {
        if (clientState.compareAndSet(ConnectionStateEnum.Shutdown, ConnectionStateEnum.Starting)) {
            connect(null);
        }
    }

    @Override
    public void stop(Listener listener) {
        if (!clientState.compareAndSet(ConnectionStateEnum.Started, ConnectionStateEnum.Shutdown)) {
            if (listener != null) {
                listener.onFailure(new ServiceException("sub links was already shutdown."));
            }
            log.error("{} was already shutdown.", this.getClass().getSimpleName());
            return;
        }
        try {
            if (channel != null) {
                channel.closeFuture().sync();
            }
        } catch (Exception e) {
            e.printStackTrace();
        }

        if (workerGroup != null) {
            workerGroup.shutdownGracefully();
        }
        log.info("netty client [{}] stopped.", this.getClass().getSimpleName());
        listener.onSuccess();
    }

    protected void initPipeline(ChannelPipeline pipeline) {
        pipeline.addLast("decoder", getDecoder());
        pipeline.addLast("encoder", getEncoder());
        pipeline.addLast("handler", getChannelHandler());
    }

    protected ChannelHandler getDecoder() {
        return new JT809MessageDecoder();
    }

    protected ChannelHandler getEncoder() {
        return new JT809MessageEncoder();
    }

    protected ChannelHandler getChannelHandler() {
        return new SubLinkHandler();
    }

    protected int getWorkThreadNum() {
        return 1;
    }
}
